// $Id: session.c 111 2011-11-19 06:33:56Z cedric.shih@gmail.com $
/*
 * Copyright 2011 Cedric Shih (cedric dot shih at gmail dot com)
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation; either version 2.1 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 */

#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <time.h>
#include <netinet/tcp.h>

#include <zcbuffer.h>

#include "codec/headers.h"
#include "codec/request.h"
#include "log.h"
#include "request.h"
#include "server.h"
#include "sessions.h"
#include "status.h"
#include "track.h"
#include "media.h"

#include "session.h"

#define MIN(a, b) a > b ? b : a

int niortsp_session_pending(struct niortsp_session *session);
int niortsp_session_enable_read(struct niortsp_session *session);
void niortsp_session_read_cb(int fd, short event, void *arg);
int niortsp_session_enable_write(struct niortsp_session *session);
void niortsp_session_write_cb(int fd, short event, void *arg);
int niortsp_session_dispatch(struct niortsp_session *session,
		struct niortsp_request *req);

int niortsp_session_start(struct niortsp_session *session,
		struct sockaddr_in *remote)
{
	int rc;

	niortsp_debug("starting: %s:%d", inet_ntoa(remote->sin_addr),
			ntohs(remote->sin_port));

	niortsp_trace("initializing: %p", &session->read);

	event_set(&session->read, session->fd, EV_READ,
			niortsp_session_read_cb, session);

	if (event_base_set(session->server->evbase, &session->read)) {
		rc = errno ? errno : -1;
		niortsp_error("failed to assign event base: %s", strerror(rc));
		goto finally;
	}

	niortsp_trace("initializing: %p", &session->write);

	event_set(&session->write, session->fd, EV_WRITE,
			niortsp_session_write_cb, session);

	if (event_base_set(session->server->evbase, &session->write)) {
		rc = errno ? errno : -1;
		niortsp_error("failed to assign event base: %s", strerror(rc));
		goto finally;
	}

	if ((rc = zcbuffer_set_limit(session->output_buffer, 0))) {
		niortsp_error("failed to set limit: %s", strerror(rc));
		goto finally;
	}

	if ((rc = niortsp_request_decoder_reset(session->decoder,
			session->request_buffer))) {
		niortsp_error("failed to reset decoder: %s", strerror(rc));
		goto finally;
	}

	if ((rc = niortsp_session_enable_read(session))) {
		niortsp_error("failed to enable read: %s", strerror(rc));
		goto finally;
	}

	memcpy(&session->remote, remote, sizeof(session->remote));
	rc = 0;

finally:
	return rc;
}

int niortsp_session_close(struct niortsp_session *session)
{
	int rc;
	struct niortsp_stream *stream;

	if (event_initialized(&session->read) &&
			event_pending(&session->read, EV_READ, NULL)) {
		if ((rc = event_del(&session->read))) {
			niortsp_error("failed to delete event: %s",
					strerror(rc));
			goto finally;
		}
	}

	if (event_initialized(&session->write) &&
			event_pending(&session->write, EV_WRITE, NULL)) {
		if ((rc = event_del(&session->write))) {
			niortsp_error("failed to delete event: %s",
					strerror(rc));
			goto finally;
		}
	}

	if (event_initialized(&session->timer) &&
			evtimer_pending(&session->timer, NULL)) {
		if ((rc = evtimer_del(&session->timer))) {
			niortsp_error("failed to delete timer: %s",
					strerror(rc));
			goto finally;
		}
	}

	TAILQ_FOREACH(stream, &session->streams, entry) {
		niortsp_debug("invoking %p to teardown: %p",
				stream->transport->teardown,
				stream->transporter);

		if ((rc = (*stream->transport->teardown)(
				stream->transporter))) {
			niortsp_error("failed to teardown: %s", strerror(rc));
			rc = EINVAL;
			goto finally;
		}

		niortsp_debug("invoking %p to destroy: %p",
				stream->transport->destroy,
				stream->transporter);

		(*stream->transport->destroy)(
				stream->transporter);

		stream->transport = NULL;
		stream->transporter = NULL;

		niortsp_debug("unlistening track: %s", stream->track->name);

		if ((rc = niortsp_track_unlisten(stream->track,
				niortsp_session_track_cb, stream))) {
			niortsp_error("failed to unlisten track: %s",
					strerror(rc));
			goto finally;
		}

		stream->track = NULL;
	}

	niortsp_debug("closing socket: %d", session->fd);

	if ((rc = close(session->fd))) {
		rc = errno ? errno : -1;
		niortsp_error("failed to close: %s", strerror(rc));
		goto finally;
	}

	session->fd = -1;

	niortsp_sessions_remove(session->server->sessions, session);
	rc = 0;

finally:
	return rc;
}

int niortsp_session_read_pending(struct niortsp_session *session)
{
	if (event_initialized(&session->read) &&
			event_pending(&session->read, EV_READ, NULL)) {
		return 1;
	}

	return 0;
}

int niortsp_session_write_pending(struct niortsp_session *session)
{
	if (event_initialized(&session->write) &&
			event_pending(&session->write, EV_WRITE, NULL)) {
		return 1;
	}

	return 0;
}

int niortsp_session_pending(struct niortsp_session *session)
{
	return niortsp_session_read_pending(session) |
			niortsp_session_write_pending(session);
}

int niortsp_session_enable_read(struct niortsp_session *session)
{
	int rc;
	struct timeval *tv;

	niortsp_trace("waiting for socket readable: %d", session->fd);

	if (niortsp_session_pending(session)) {
		rc = EINPROGRESS;
		niortsp_error("I/O pending");
		goto finally;
	}

	if (session->server->timeout.read > 0) {
		session->timeout.tv_sec = session->server->timeout.read;
		session->timeout.tv_usec = 0;
		tv = &session->timeout;
	} else {
		tv = NULL;
	}

	niortsp_trace("enabling: %p", &session->read);

	if (event_add(&session->read, tv)) {
		rc = errno ? errno : -1;
		niortsp_error("failed to add event for write: %s", strerror(rc));
		goto finally;
	}

	rc = 0;

finally:
	return rc;
}

void niortsp_session_read_cb(int fd, short event, void *arg)
{
	int rc, b;
	ssize_t n;
	struct niortsp_session *session = arg;
	struct zcbuffer *buffer;
	struct niortsp_request *req;
	unsigned char header[4];

	if (event & EV_TIMEOUT) {
		rc = ETIMEDOUT;
		niortsp_error("timed out: %d", event);
		goto finally;
	}

	if (!(event & EV_READ)) {
		rc = EIO;
		niortsp_error("not readable: %d", event);
		goto finally;
	}

	niortsp_trace("reading data from socket %d", fd);

	buffer = session->input_buffer;

	if ((n = zcbuffer_read(buffer, session->fd)) <= 0) {
		if (n == 0) {
			niortsp_debug("connection reset");
			rc = ECONNRESET;
			goto finally;
		} else {
			rc = errno ? errno : -1;

			if (rc != EAGAIN && rc != EINPROGRESS) {
				niortsp_error("failed to read: %s", strerror(rc));
				goto finally;
			}
		}
	}

	zcbuffer_flip(buffer);

	while (zcbuffer_remaining(buffer) > 0) {
		niortsp_trace("processing remaining buffer: %d bytes",
				zcbuffer_remaining(buffer));

		if ((rc = zcbuffer_peek(buffer, zcbuffer_position(buffer),
				&b))) {
			niortsp_error("failed to peek: %s", strerror(rc));
			goto finally;
		}

		if (b == '$') {
			if (zcbuffer_remaining(buffer) < sizeof(header)) {
				niortsp_debug("waiting for next chunk of data");
				break;
			}

			niortsp_trace("processing interleaved header");

			if ((rc = zcbuffer_peeks(buffer,
					zcbuffer_position(buffer),
					header, sizeof(header)))) {
				niortsp_error("failed to peek header: %s",
						strerror(rc));
				goto finally;
			}

			if (header[1] > session->num_streams) {
				rc = EPROTO;
				niortsp_error("invalid stream id: %d",
						header[1]);
				goto finally;
			}

			n = ntohs(*((u_int16_t *)(header + 2)));

			if (zcbuffer_remaining(buffer) < (n + sizeof(header))) {
				niortsp_debug("more interleaved to read: "
						"%d bytes", n + sizeof(header) -
						zcbuffer_remaining(buffer));
				break;
			}

			niortsp_warn("skipping incoming payload: %d bytes",
					n + sizeof(header));

			zcbuffer_skip(buffer, n + sizeof(header));
			continue;
		}

		if ((rc = niortsp_request_decoder_parse(session->decoder,
				buffer, &req))) {
			niortsp_error("failed to parse: %s", strerror(rc));
			goto finally;
		}

		if (!req) {
			niortsp_debug("waiting for next chunk of data");
			break;
		}

		if (strcmp("RTSP", niortsp_request_get_protocol(req))) {
			niortsp_error("unsupported request: %s %s",
					niortsp_request_get_protocol(req),
					niortsp_request_get_uri(req));
			if ((rc = niortsp_session_reply(
					session, req, 400) != 0)) {
				niortsp_error("failed to reply: %s",
						strerror(rc));
				goto finally;
			}

			continue;
		}

		if ((rc = niortsp_session_dispatch(session, req))) {
			niortsp_error("failed to dispatch: %s",
					strerror(rc));
			goto finally;
		}

		if ((rc = niortsp_request_decoder_reset(session->decoder,
				session->request_buffer))) {
			niortsp_error("failed to reset decoder: %s",
					strerror(rc));
			goto finally;
		}
	}

	if ((rc = zcbuffer_compact(buffer))) {
		niortsp_error("failed to compact: %s", strerror(rc));
		goto finally;
	}

	if (!niortsp_session_pending(session)) {
		if ((rc = niortsp_session_enable_read(session))) {
			niortsp_error("failed to enable read: %s",strerror(rc));
			goto finally;
		}
	}

	rc = 0;

finally:
	if (rc != 0) {
		niortsp_debug("connection error: %s:%d (%s)",
				inet_ntoa(session->remote.sin_addr),
				ntohs(session->remote.sin_port), strerror(rc));
		niortsp_session_close(session);
	}
	return;
}

int niortsp_session_enable_write(struct niortsp_session *session)
{
	int rc;
	struct timeval *tv;

	if (niortsp_session_pending(session)) {
		rc = EINPROGRESS;
		niortsp_error("I/O pending");
		goto finally;
	}

	niortsp_trace("waiting for socket writable: %d", session->fd);

	if (session->server->timeout.write > 0) {
		session->timeout.tv_sec = session->server->timeout.write;
		session->timeout.tv_usec = 0;
		tv = &session->timeout;
	} else {
		tv = NULL;
	}

	if (event_add(&session->write, tv)) {
		rc = errno ? errno : -1;
		niortsp_error("failed to add event for write: %s", strerror(rc));
		goto finally;
	}

	rc = 0;

finally:
	return rc;
}

inline int niortsp_session_control_pending(struct niortsp_session *session)
{
	return zcbuffer_remaining(session->output_buffer) > 0;
}

inline struct niortsp_stream *niortsp_session_pending_stream(
		struct niortsp_session *session)
{
	struct niortsp_stream *stream;

	TAILQ_FOREACH(stream, &session->streams, entry) {
		niortsp_debug("invoking %p to determine pending: %p",
				stream->transport->pending,
				stream->transporter);

		if ((*stream->transport->pending)(stream->transporter)) {
			return stream;
		}
	}

	return stream;
}

inline int niortsp_session_output_pending(
		struct niortsp_session *session)
{
	return niortsp_session_control_pending(session) ||
			niortsp_session_pending_stream(session);
}

int niortsp_session_write(struct niortsp_session *session)
{
	int rc, flag = 1, corked = 0;
	struct niortsp_stream *stream;

	niortsp_trace("turning on TCP_CORK");

	if (setsockopt(session->fd, IPPROTO_TCP, TCP_CORK,
			(char *)&flag, sizeof(flag)) == -1) {
		rc = errno ? errno : -1;
		niortsp_error("failed to turn on TCP_CORK: %s",
				strerror(rc));
		goto finally;
	}

	corked = 1;

	niortsp_trace("writing pending output");

	if (niortsp_session_control_pending(session)) {
		if ((zcbuffer_write(session->output_buffer,
				session->fd)) <= 0) {
			rc = errno ? errno : -1;

			if (rc != EAGAIN && rc != EINPROGRESS) {
				niortsp_error("failed to write control data: %s",
						strerror(rc));
			}

			goto finally;
		}

		if ((rc = zcbuffer_compact(session->output_buffer))) {
			niortsp_error("failed to compact: %s", strerror(rc));
			goto finally;
		}

		zcbuffer_flip(session->output_buffer);
	} else if ((stream = niortsp_session_pending_stream(session))) {
		niortsp_debug("invoking %p to transport video: %p",
				stream->transport->next, stream->transporter);

		if ((rc = (*stream->transport->next)( stream->transporter))) {
			if (rc != EAGAIN && rc != EINPROGRESS) {
				niortsp_error("failed to transport video: %s",
						strerror(rc));
			}

			goto finally;
		}
	}

	rc = 0;

finally:
	if (corked) {
		niortsp_trace("turning off TCP_CORK");

		flag = 0;

		if (setsockopt(session->fd, IPPROTO_TCP, TCP_CORK,
				(char *)&flag, sizeof(flag)) == -1) {
			niortsp_error("failed to turn off TCP_CORK: %s",
					strerror(errno));
		}
	}
	return rc;
}

void niortsp_session_write_cb(int fd, short event, void *arg)
{
	int rc;
	struct niortsp_session *session = arg;

	if (event & EV_TIMEOUT) {
		rc = ETIMEDOUT;
		niortsp_error("timed out: %d", event);
		goto finally;
	}

	if (!(event & EV_WRITE)) {
		rc = EIO;
		niortsp_error("not writable: %d", event);
		goto finally;
	}

	niortsp_trace("writing data to socket %d", fd);

	while (niortsp_session_output_pending(session)) {
		if ((rc = niortsp_session_write(session))) {
			if (rc != EAGAIN && rc != EINPROGRESS) {
				niortsp_error("failed to write: %s", strerror(rc));
				goto finally;
			}

			niortsp_warn("transmission in progress; potential slow "
					"connection: %s:%d",
					inet_ntoa(session->remote.sin_addr),
					ntohs(session->remote.sin_port));
			break;
		}
	}

	if (niortsp_session_output_pending(session)) {
		if ((rc = niortsp_session_enable_write(session))) {
			niortsp_error("failed to enable write: %s",
					strerror(rc));
			goto finally;
		}
	} else {
		if (session->need_close) {
			niortsp_debug("closing session");

			if ((rc = niortsp_session_close(session))) {
				niortsp_error("failed to close session: %s",
						strerror(rc));
				goto finally;
			}
		} else {
			if ((rc = niortsp_session_enable_read(session))) {
				niortsp_error("failed to enable read: %s",
						strerror(rc));
				goto finally;
			}
		}
	}

	rc = 0;

finally:
	if (rc != 0) {
		niortsp_debug("connection error: %s:%d (%s)",
				inet_ntoa(session->remote.sin_addr),
				ntohs(session->remote.sin_port),
				strerror(rc));
		niortsp_session_close(session);
	}
	return;
}

int niortsp_session_get_int_header(struct niortsp_request *req,
		const char *name, int *value)
{
	int rc;
	char *ptr;
	long int val;
	const char *header;

	if (!(header = niortsp_request_get_input_header(req, name))) {
		niortsp_error("missing '%s' header", name);
		rc = ENOENT;
		goto finally;
	}

	errno = 0;
	val = strtol(header, &ptr, 10);

	if (errno != 0) {
		rc = errno;
		niortsp_error("failed to convert '%s' header '%s' to integer: %s",
				name, header, strerror(rc));
		goto finally;
	} else if (*ptr != '\0') {
		niortsp_error("invalid integer format of '%s' header: %s",
				name, header);
		rc = EINVAL;
		goto finally;
	}

	*value = val;
	rc = 0;
finally:
	return rc;
}

int niortsp_session_dispatch(struct niortsp_session *session,
		struct niortsp_request *req)
{
	int rc, i;
	struct niortsp_handler_entry *handler;

	niortsp_debug("handling request: %s %s", req->method, req->uri);

	if ((rc = niortsp_session_get_int_header(req, "CSeq", &i))) {
		niortsp_error("invalid sequence number");

		if ((rc = niortsp_session_reply(session, req, 400) != 0)) {
			niortsp_error("failed to reply: %s", strerror(rc));
			goto finally;
		}

		rc = 0;
		goto finally;
	}

	if (session->sequence == 0) {
		// tolerate Live555 with multiple clients
		session->sequence = i;
	} else {
		session->sequence++;

		if (i != session->sequence) {
			niortsp_error("mismatched sequence number: %d != %d",
					i, session->sequence);

			if ((rc = niortsp_session_reply(
					session, req, 400) != 0)) {
				niortsp_error("failed to reply: %s",
						strerror(rc));
				goto finally;
			}

			rc = 0;
			goto finally;
		}
	}

	if (req->version.major != 1 || req->version.minor != 0) {
		niortsp_error("protocol version not supported: %d.%d",
				req->version.major, req->version.minor);

		if ((rc = niortsp_session_reply(session, req, 505) != 0)) {
			niortsp_error("failed to reply: %s", strerror(rc));
			goto finally;
		}

		rc = 0;
		goto finally;
	}

	niortsp_debug("looking for handler: %s", req->method);

	if (!(handler = niortsp_handlers_find(session->server->handlers,
			req->method))) {
		niortsp_error("not implemented: %s", req->method);

		if ((rc = niortsp_session_reply(
				session, req, 501) != 0)) {
			niortsp_error("failed to reply: %s", strerror(rc));
			goto finally;
		}

		rc = 0;
		goto finally;
	}

	niortsp_debug("invoking handler: %s %s", req->method, req->uri);

	if ((rc = (*(handler->cb))(session, req, handler->cbarg))) {
		niortsp_error("failed to handle method: %s %s",
				req->method, req->uri);

		if ((rc = niortsp_session_reply(session, req, 500) != 0)) {
			niortsp_error("failed to reply: %s", strerror(rc));
			goto finally;
		}

		rc = 0;
		goto finally;
	}

	rc = 0;

finally:
	if (rc != 0) {
		niortsp_session_close(session);
	}
	return rc;
}

int niortsp_session_reply(struct niortsp_session *session,
		struct niortsp_request *req, int code)
{
	int rc, i;
	const char *header, *reason = NULL;
	char temp[64];
	struct tm cur;
	struct tm *cur_p;
	struct zcbuffer *buffer;

	niortsp_debug("sending reply: %d", code);

	for (i = 0; i < num_status; i++) {
		if (status[i].code == code) {
			reason = status[i].reason;
			break;
		}
	}

	if (!reason) {
		niortsp_error("status code not defined: %d", code);
		rc = -1;
		goto finally;
	}

	if (((header = niortsp_request_get_output_header(
			req, "Connection")) && !strcasecmp(header, "close")) ||
			(code != 200)) {
		session->need_close = 1;
	}

	if (!header && session->need_close) {
		niortsp_debug("adding 'Connection' header: %d",
				session->sequence);

		if ((rc = niortsp_request_set_output_header(req,
				"Connection", "close"))) {
			niortsp_error("failed to add connection header: %s",
					strerror(rc));
			goto finally;
		}
	}

	if (!niortsp_request_get_output_header(req, "CSeq")) {
		niortsp_debug("adding 'CSeq' header: %d", session->sequence);

		snprintf(temp, sizeof(temp), "%d", session->sequence);

		if ((rc = niortsp_request_set_output_header(req,
				"CSeq", temp))) {
			niortsp_error("failed to add sequence header: %s",
					strerror(rc));
			goto finally;
		}
	}

	if (!niortsp_request_get_output_header(req, "Date")) {
		niortsp_debug("adding 'Date' header: %d", session->sequence);

		time_t t = time(NULL);
		gmtime_r(&t, &cur);
		cur_p = &cur;

		if (strftime(temp, sizeof(temp),
			"%a, %d %b %Y %H:%M:%S GMT", cur_p) <= 0) {
			rc = errno ? errno : -1;
			niortsp_error("failed to format date: %s", strerror(rc));
			goto finally;
		}

		if ((rc = niortsp_request_set_output_header(req,
				"Date", temp))) {
			niortsp_error("failed to add date header: %s",
					strerror(rc));
			goto finally;
		}
	}

	buffer = niortsp_request_get_buffer(req);

	niortsp_debug("flipping request buffer for read");

	zcbuffer_flip(buffer);

	if (!niortsp_request_get_output_header(req, "Content-Length") &&
			zcbuffer_remaining(buffer) > 0) {
		niortsp_debug("adding 'Content-Length' header: %d",
				zcbuffer_remaining(buffer));

		snprintf(temp, sizeof(temp), "%d", zcbuffer_remaining(buffer));

		if ((rc = niortsp_request_set_output_header(req,
				"Content-Length", temp))) {
			niortsp_error("failed to add content length header: %s",
					strerror(rc));
			goto finally;
		}
	}

	niortsp_debug("preparing output: %d %s", code, reason);

	if ((rc = zcbuffer_set_position(session->output_buffer,
			zcbuffer_limit(session->output_buffer)))) {
		niortsp_error("failed to set position: %s", strerror(rc));
		goto finally;
	}

	if ((rc = zcbuffer_set_limit(session->output_buffer,
			zcbuffer_capacity(session->output_buffer)))) {
		niortsp_error("failed to set limit: %s", strerror(rc));
		goto finally;
	}

	if ((rc = zcbuffer_putf(session->output_buffer, "%s/%d.%d %d %s\r\n",
			req->protocol, req->version.major, req->version.minor,
			code, reason))) {
		niortsp_error("failed to prepare first line: %s", strerror(rc));
		goto finally;
	}

	if ((rc = niortsp_headers_format(req->output_headers,
			session->output_buffer))) {
		niortsp_error("failed to format headers: %s", strerror(rc));
		goto finally;
	}

	if (zcbuffer_remaining(buffer) > 0 && (rc = zcbuffer_load(
			session->output_buffer, buffer))) {
		rc = errno ? errno : -1;
		niortsp_error("failed to load: %s", strerror(rc));
		goto finally;
	}

	zcbuffer_flip(session->output_buffer);

	if ((rc = niortsp_session_enable_write(session))) {
		niortsp_error("failed to enable write: %s", strerror(rc));
		goto finally;
	}

	rc = 0;

finally:
	return rc;
}

int niortsp_session_setup(struct niortsp_session *session,
		struct niortsp_track *track, enum niortp_transport_type type,
		struct niortp_transport *transport, unsigned int mtu,
		unsigned short rtp, unsigned short rtcp)
{
	int rc, fd, setup = 0;
	struct niortp_transport_param param;
	struct niortsp_stream *stream = NULL;
	void *transporter = NULL;

	if (!session) {
		niortsp_error("null argument(s)");
		rc = EINVAL;
		goto finally;
	}

	if (niortsp_stream_list_find(&session->streams, track)) {
		niortsp_error("already setup: %s/%s", track->media->name,
				track->name);
		rc = EEXIST;
		goto finally;
	}

	niortsp_debug("setting up: %s/%s", track->media->name, track->name);

	memset(&param, '\0', sizeof(param));

	switch (type) {
	case NIORTP_TRANSPORT_UDP_UNICAST:
		memcpy(&param.address.rtp, &session->remote,
				sizeof(param.address.rtp));
		memcpy(&param.address.rtcp, &session->remote,
				sizeof(param.address.rtcp));
	case NIORTP_TRANSPORT_INTERLEAVED:
		param.type = type;
		param.address.rtp.sin_port = htons(rtp);
		param.address.rtcp.sin_port = htons(rtcp);
		break;
	default:
		niortsp_error("unknown transport type: %d", type);
		rc = EINVAL;
		goto finally;
	}

	memcpy(&param.remote, &session->remote, sizeof(param.remote));
	param.type = type;
	param.payload_type = track->payload_type;
	param.mtu = mtu;

	fd = param.type == NIORTP_TRANSPORT_INTERLEAVED ?
			session->fd : session->server->rtp.fd;

	niortsp_debug("invoking %p to create", transport->create);

	if (!(transporter = (*transport->create)(session->server->evbase))) {
		rc = errno ? errno : -1;
		niortsp_error("failed to create transport: %s",
				strerror(rc));
		goto finally;
	}

	niortsp_debug("invoking %p to setup: %p", transport->setup,
			transporter);

	if ((rc = (*transport->setup)(transporter, fd, &param))) {
		niortsp_error("failed to setup: %s", strerror(rc));
		rc = EINVAL;
		goto finally;
	}

	setup = 1;

	if ((rc = niortsp_stream_list_add(&session->streams, session, track,
			&param, transport, transporter, &stream))) {
		niortsp_error("failed to add: %s", strerror(rc));
		goto finally;
	}

	niortsp_debug("listening source: %s", track->name);

	if ((rc = niortsp_track_listen(track, niortsp_session_track_cb,
			stream))) {
		niortsp_error("failed to listen: %s", strerror(rc));
		goto finally;
	}

	rc = 0;

finally:
	if (rc != 0) {
		if (stream) {
			niortsp_stream_list_del(&session->streams, stream);
		}

		if (transport) {
			if (setup) {
				niortsp_debug("invoking %p to teardown: %p",
						transport->setup, transporter);

				(*transport->teardown)(transporter);
			}

			niortsp_debug("invoking %p to destroy: %p",
					transport->setup, transporter);

			(*transport->destroy)(transporter);
		}
	}
	return rc;
}

void niortsp_session_stat_cb(int fd, short event, void *arg)
{
	int rc;
	struct niortp_transport_stat stat;
	struct niortsp_session *session = arg;
	struct niortsp_stream *stream;

	TAILQ_FOREACH(stream, &session->streams, entry) {
		niortsp_debug("invoking %p to stat: %p",
				stream->transport->stat, stream->transporter);

		if ((rc = (*stream->transport->stat)(
				stream->transporter, &stat))) {
			niortsp_error("failed to stat: %s", strerror(rc));
			rc = EINVAL;
			goto finally;
		}

		fprintf(stderr, "track=%s/%s, "
				"client=%s:%d, payload=%d, transport=%s, "
				"fps=%d, in=%d/%d, drop=%d/%d, out=%d/%d, "
				"duration=[%d,%d], rtp=%d/%d, rtcp=%d/%d\n",
				stream->track->media->name,
				stream->track->name,
				inet_ntoa(stream->param.remote.sin_addr),
				ntohs(stream->param.remote.sin_port),
				stream->track->payload_type,
				niortp_transport_type_to_str(stream->param.type),
				stat.instances,
				stat.incoming.units,
				stat.incoming.bytes,
				stat.dropped.units,
				stat.dropped.bytes,
				stat.outgoing.units,
				stat.outgoing.bytes,
				stat.duration.min,
				stat.duration.max,
				stat.rtp.units,
				stat.rtp.bytes,
				stat.rtcp.units,
				stat.rtcp.bytes);
	}

finally:
	if (event_add(&session->timer, &session->duration)) {
		rc = errno ? errno : -1;
		niortsp_error("failed to add timer: %s", strerror(rc));
	}
	return;
}

int niortsp_session_pause(struct niortsp_session *session)
{
	int rc;
	struct niortsp_stream *stream;

	if (!session) {
		rc = EINVAL;
		niortsp_error("null pointer(s)");
		goto finally;
	}

	TAILQ_FOREACH(stream, &session->streams, entry) {
		niortsp_debug("pausing track: %s", stream->track->name);

		stream->playing = 0;
	}

	rc = 0;

finally:
	return rc;
}

int niortsp_session_play(struct niortsp_session *session)
{
	int rc;
	struct niortsp_stream *stream;

	if (evtimer_initialized(&session->timer) &&
			evtimer_pending(&session->timer, NULL)) {
		rc = EINPROGRESS;
		niortsp_error("already playing");
		goto finally;
	}

	niortsp_debug("playing: %s:%d", inet_ntoa(session->remote.sin_addr),
			ntohs(session->remote.sin_port));

	evtimer_set(&session->timer, niortsp_session_stat_cb, session);

	if (event_base_set(session->server->evbase, &session->timer)) {
		rc = errno ? errno : -1;
		niortsp_error("failed to assign event base: %s", strerror(rc));
		goto finally;
	}

	session->duration.tv_sec = 1;
	session->duration.tv_usec = 0;

	if (event_add(&session->timer, &session->duration)) {
		rc = errno ? errno : -1;
		niortsp_error("failed to add timer: %s", strerror(rc));
		goto finally;
	}

	TAILQ_FOREACH(stream, &session->streams, entry) {
		stream->playing = 1;

		niortsp_debug("refreshing track: %s", stream->track->name);

		if ((rc = niortsp_track_refresh(stream->track))) {
			niortsp_error("failed to refresh: %s", strerror(rc));
			goto finally;
		}
	}

	rc = 0;

finally:
	return rc;
}

void niortsp_session_track_cb(struct timeval *timestamp,
		struct zcbuffer *data, void *cbarg)
{
	int rc;
	struct niortsp_stream *stream = cbarg;

	niortsp_trace("processing data: %p", data);

	if (!stream->playing) {
		rc = 0;
		goto finally;
	}

	if (niortsp_session_control_pending(stream->session)) {
		niortsp_debug("control session in progress");
		rc = 0;
		goto finally;
	}

	niortsp_debug("invoking %p to send: %p", stream->transport->send,
			stream->transporter);

	if ((rc = (*stream->transport->send)(stream->transporter,
			timestamp, data))) {
		niortsp_error("failed to setup: %s", strerror(rc));
		rc = EINVAL;
		goto finally;
	}

	if (niortsp_session_pending_stream(stream->session)) {
		if (niortsp_session_read_pending(stream->session)) {
			if ((rc = event_del(&stream->session->read))) {
				rc = errno ? errno : -1;
				niortsp_error("failed to delete: %s",
						strerror(rc));
				goto finally;
			}
		}

		if (!niortsp_session_write_pending(stream->session)) {
			if ((rc = niortsp_session_enable_write(
					stream->session))) {
				niortsp_error("failed to enable write: %s",
						strerror(rc));
				goto finally;
			}
		}
	}

	rc = 0;

finally:
	if (rc != 0) {
		niortsp_debug("data error: %s:%d (%s)",
				inet_ntoa(stream->session->remote.sin_addr),
				ntohs(stream->session->remote.sin_port),
				strerror(rc));
		niortsp_session_close(stream->session);
	}
	return;
}
